package mapReduce;

import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;

import vo.Comment;

public class ListCommentsMR {

	public static class Mapper extends
			TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {
		
		protected void map(ImmutableBytesWritable key, Result result,
				Context context) throws IOException, InterruptedException {

			String text = Bytes.toString(result
					.getRow());

			ImmutableBytesWritable immutableKeyComment = new ImmutableBytesWritable(
					Bytes.toBytes(text));
			
			ImmutableBytesWritable immutableValueVote = new ImmutableBytesWritable(result.getValue(Bytes.toBytes(Comment.CF_ATTRIBUTES), 
					Bytes.toBytes(Comment.COMMENT_TEXT)));
			context.write(immutableKeyComment, immutableValueVote);
		}

	}

	public static class Reducer
			extends
			TableReducer<ImmutableBytesWritable, ImmutableBytesWritable, NullWritable> {

		private static Map<String, String> map = new HashMap<String, String>();

		protected void reduce(ImmutableBytesWritable key,
				Iterable<ImmutableBytesWritable> it, Context context)
				throws IOException, InterruptedException {

			String word = Bytes.toString(key.get());
			String valorVote = "";
			for (ImmutableBytesWritable immutableBytesWritable : it) {
				valorVote = Bytes.toString(immutableBytesWritable.get());
			}
			map.put(word, valorVote);
		}

		public static String printListComments() {
			List<Map.Entry<String, String>> list = new LinkedList<Map.Entry<String, String>>(
					map.entrySet());
			return list.toString();
		}

	}
}
